package com.icbc.dccsh.bolt;

import java.util.HashMap;
import java.util.Map;

import org.apache.commons.lang.StringUtils;

import com.icbc.dccsh.util.Watcher;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;

public class CounterBolt extends BaseBasicBolt {
	private long now = 0;
	private Watcher _watcher;
	private String _mockName;

	public CounterBolt() {

	}

	public CounterBolt(String mockName) {
		this._mockName = mockName;
	}

	@Override
	public void prepare(Map stormConf, TopologyContext context) {
		now = System.currentTimeMillis();
		String clazzName = this._mockName;
		if (StringUtils.isEmpty(clazzName)) {
			this._watcher = Watcher.getWatcher(this.getClass());
		} else {
			this._watcher = Watcher.getWatcher(clazzName);
		}
	}

	private Integer counter = 0;

	public void execute(Tuple tuple, BasicOutputCollector collector) {
		if (this._mockName.equals("formula")){
//			Utils.sleep(5);
//			System.out.println("sleeping...");
		}
		this._watcher.count();
		counter++;
		Object o = tuple.getValue(0);
		if (counter % 100 == 0) {
			long delta = System.currentTimeMillis() - now;
			System.out.println(o);
			// System.out.println(counter);
			System.out.println(String.format("%d, time elapsed... %d ms", counter, delta));
		}
		collector.emit(new Values(o));
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		declarer.declare(new Fields("count"));
	}

	@Override
	public void cleanup() {
		System.out.println("Total count is " + counter.toString());
	}
}